package spring.cloud.tasks.executor.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.utils.CloseableExecutorService;
import spring.cloud.tasks.executor.node.TaskNodePaths;
import spring.cloud.tasks.executor.utils.ExecutorServiceUtil;
import spring.cloud.tasks.executor.utils.ThreadFactory;
import spring.cloud.tasks.executor.utils.log.LogEvents;
import spring.cloud.tasks.executor.utils.log.LogUtils;
import spring.cloud.tasks.executor.utils.zookeeper.Zookeeper;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;

@Slf4j
@Setter
@Getter
public class TaskInitService {

    /**
     * record the alarm message hashcode, permanently saved, used for just raising alarm one time for one type exception
     */
    public static final ConcurrentMap<String, ConcurrentMap<String, Set<Integer>>> JOB_INIT_FAILED_RECORDS = new ConcurrentHashMap<>();
    public Zookeeper zookeeper;
    public String executorId;
    public TaskInitService taskInitService;
    public TreeCache treeCache;
    public java.util.concurrent.ExecutorService executorService;
    public List<String> taskIdSet = new ArrayList<>();

    public TaskInitService(ExecutorAndTaskInitService executorAndTaskInitService) {
        this.zookeeper = executorAndTaskInitService.getZookeeper();
        this.executorId = executorAndTaskInitService.getExecutorId();
        this.taskInitService = executorAndTaskInitService.getTaskInitService();
        JOB_INIT_FAILED_RECORDS.putIfAbsent(executorId, new ConcurrentHashMap<String, Set<Integer>>());
    }

    public boolean removeTaskId(String taskId) {
        return taskIdSet.remove(taskId);
    }

    public void start() throws Exception {
        treeCache = TreeCache.newBuilder(zookeeper.getCuratorFramework(), TaskNodePaths.ROOT).setExecutor(new CloseableExecutorService(Executors.newSingleThreadExecutor(new ThreadFactory(executorId + "-$Jobs-watcher", false)), true)).setMaxDepth(1).build();
        executorService = Executors.newSingleThreadExecutor(new ThreadFactory(executorId + "-initNewJob-thread", false));
        treeCache.getListenable().addListener(new TaskInitListener(this), executorService);
        treeCache.start();
    }

    public void shutdown() {
        try {
            if (treeCache != null) {
                treeCache.close();
            }
        } catch (Throwable t) {
            LogUtils.error(log, LogEvents.ExecutorEvent.INIT_OR_SHUTDOWN, t.toString(), t);
        }
        try {
            ExecutorServiceUtil.shutdownNow(executorService);
        } catch (Throwable t) {
            LogUtils.error(log, LogEvents.ExecutorEvent.INIT_OR_SHUTDOWN, t.toString(), t);
        }
    }


}
